package cn.ipanel.bigdata.job.basic

import cn.ipanel.bigdata.boot.Job
import cn.ipanel.bigdata.boot.config.{ConfigKey, Configuration}
import cn.ipanel.bigdata.dw.dim.phoenix.{T_CLASS_TAG => TG, T_DEVICE => S}
import cn.ipanel.bigdata.dw.dws.{T_DEVICE_TAG => T}
import cn.ipanel.bigdata.utils.Dictionary.{F_DATE, F_STATUS, BusState}
import org.apache.spark.sql.functions.{col, count, lit, substring}


/**
 * 按设备类型 Tag 统计，目前为止 Tag 下的 class 所有分类有：
 * 'istickiot','Multifunctionwatch','ac','sensor','curtain','camera','hematometer','glucometer','ifree','igot','oximeter','itee','rowmachine','light','safetybeeper','skiprope','scale','smartbed','thermostat','thermometer','tv','uricacidmeter','windowpusher'
 */
class DeviceByTag extends Job {

  val F_TAG = T.F_TAG

  override def onStartup(): Unit = {
    calByDay()
    calByMonth()
  }


  def calByDay(): Unit = {
    T.deleteByPeriodTime(_DAY.toSimpleDate)

    T.save(
      S.load
        .filter(
          substring(col(S.F_REGION), 1, 6) === lit(Configuration.getParam(ConfigKey.ROOT_AREA_CODE).substring(0, 6))
          and  (
            // 截止到当天的设备总数，只需创建时间小于当天，且未删除；或状态是删除，但是删除时间在当天之后
            (col(S.F_CREATE_TIME) < lit(_DAY.next().printFormat("yyyy-MM-dd")) and col(F_STATUS) =!= lit(BusState.DEL))
            or (col(F_STATUS) === lit(BusState.DEL) and col(S.F_UPDATE_TIME) >= lit(_DAY.next().printFormat("yyyy-MM-dd")))
            )
        )
        .select(S.F_DEVICE_ID, S.F_DEVICE_CLASS, S.F_REGION)
        .join(
          // 补充 TAG 字段
          TG.load
            .filter(col(F_DATE) === lit(_DAY.toDate))
            .select(TG.F_CLASS, TG.F_TAG)
            .withColumnRenamed(TG.F_CLASS, S.F_DEVICE_CLASS)
            .withColumnRenamed(TG.F_TAG, F_TAG),
          Seq(S.F_DEVICE_CLASS)
        )
        .groupBy(S.F_REGION, F_TAG)
        .agg(
          count(S.F_DEVICE_ID) as T.F_DEVICE_COUNT
        )
        .withColumn(T.F_PERIOD_TIME, lit(_DAY.toSimpleDate))
    )
  }

  def calByMonth(): Unit = {
    val thisMonth = _DAY.asMonth
    val nextMonth = thisMonth.next()

    T.deleteByPeriodTime(_DAY.toSimpleMonth)
    T.save(
      S.load
        .filter(
          substring(col(S.F_REGION), 1, 6) === lit(Configuration.getParam(ConfigKey.ROOT_AREA_CODE).substring(0, 6))
          and (
          //   创建时间为当月最后一天之前，状态不为删除（可以编辑）；或创建时间为当月，虽然删除，但删除时间为当月之后
              (col(F_STATUS) =!= lit(BusState.DEL) and col(S.F_CREATE_TIME) < lit(nextMonth.printFormat("yyyy-MM-dd")))
              or (col(F_STATUS) === lit(BusState.DEL) and col(S.F_UPDATE_TIME) >= lit(nextMonth.printFormat("yyyy-MM-dd")))
              )
        )
        .select(S.F_DEVICE_ID, S.F_DEVICE_CLASS, S.F_REGION)
        .join(
          // 补充 TAG 字段
          TG.load
            .filter(col(F_DATE) === lit(_DAY.toDate))
            .select(TG.F_CLASS, TG.F_TAG)
            .withColumnRenamed(TG.F_CLASS, S.F_DEVICE_CLASS)
            .withColumnRenamed(TG.F_TAG, F_TAG),
          Seq(S.F_DEVICE_CLASS)
        )
        .groupBy(S.F_REGION, F_TAG)
        .agg(
          count(S.F_DEVICE_ID) as T.F_DEVICE_COUNT
        )
        .withColumn(T.F_PERIOD_TIME, lit(thisMonth.toSimpleMonth))
    )
  }


}
